package org.xiaoyu.rpc.remoting.transport.cllient.future;

import lombok.extern.slf4j.Slf4j;

import java.util.Collections;
import java.util.HashSet;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Created By Have
 * 2021/7/16 12:03
 */
@Slf4j
public class HashedWheelTimer implements Timer {
    /**
     * class
     */
    private static final AtomicInteger INSTANCE_COUNT = new AtomicInteger(0);
    private static final AtomicBoolean WARNED_TOO_MANY_INSTANCE = new AtomicBoolean(false);
    private static final int INSTANCE_COUNT_LIMIT = 16;
    private static final AtomicIntegerFieldUpdater<HashedWheelTimer> WORKER_STATE_UPDATE = AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimer.class, "workerState");

    private static final int WORKER_STATE_INIT = 0;
    private static final int WORKER_STATE_STARTED = 1;
    private static final int WORKER_STATE_SHUTDOWN = 2;

    /**
     * object
     */
    private volatile int workerState = WORKER_STATE_INIT;
    private final Worker worker = new Worker(); // 工作任务 Runnable
    private final Thread workerThread; // 工作线程，每个HashedWheelTimer对应一个工作线程

    private final long tickDuration; // 时钟周期，没tickDuration醒来一次检查timeout
    private final int mask;
    private final HashedWheelBucket[] wheel;
    private final Queue<HashedWheelTimeout> timeouts = new LinkedBlockingQueue<>(); // 超时队列 待处理
    private final Queue<HashedWheelTimeout> cancelledTimeouts= new LinkedBlockingQueue<>(); // 已经顺利完成的任务，取消timeout等待
    private final CountDownLatch startTimeInitialized = new CountDownLatch(1);
    private final AtomicLong pendingTimeouts = new AtomicLong(0); // 正在等待处理的timeout
    private final long maxPendingTimeouts;

    private volatile long startTime;

    public HashedWheelTimer() {
        this(new NamedThreadFactory());
    }

    /**
     * @param threadFactory 线程工厂
     */
    public HashedWheelTimer(NamedThreadFactory threadFactory) {
        this(threadFactory, 100, TimeUnit.MILLISECONDS);
    }

    /**
     *
     * @param tickDuration
     * @param timeUnit
     */
    public HashedWheelTimer(long tickDuration, TimeUnit timeUnit) {
        this(new NamedThreadFactory(), tickDuration, timeUnit);
    }

    public HashedWheelTimer(NamedThreadFactory threadFactory, long tickDuration, TimeUnit timeUnit) {
        this(threadFactory, tickDuration, timeUnit, 512);
    }

    /**
     *
     * @param tickDuration
     * @param timeUnit
     * @param ticksPerWheel Hashtable长度
     */
    public HashedWheelTimer(long tickDuration, TimeUnit timeUnit, int ticksPerWheel) {
        this(new NamedThreadFactory(), tickDuration, timeUnit, ticksPerWheel);
    }

    public HashedWheelTimer(
            NamedThreadFactory threadFactory,
            long tickDuration, TimeUnit timeUnit,  int ticksPerWheel) {
        this(threadFactory, tickDuration, timeUnit, ticksPerWheel, -1);
    }


    public HashedWheelTimer(
            NamedThreadFactory threadFactory,
            long tickDuration, TimeUnit timeUnit, int ticksPerWheel,
            int maxPendingTimeouts) {
        if (threadFactory == null) {
            throw new NullPointerException("HashedWheelTimer TimerThreadFactory == null");
        }
        if (timeUnit == null) {
            throw new NullPointerException("HashedWheelTimer timeUnit == null");
        }
        if (tickDuration < 0) {
            throw new IllegalArgumentException("tickDuration must be greater than 0:" + tickDuration);
        }
        if (ticksPerWheel < 0) {
            throw new IllegalArgumentException("ticksPerWheel must be greater than 0:" + tickDuration);
        }
        /**
         * HashTable
         */
        wheel = createWheel(ticksPerWheel);
        mask = wheel.length - 1;

        /**
         * 转成纳秒
         */
        this.tickDuration = timeUnit.toNanos(tickDuration);

        if (tickDuration >= Long.MAX_VALUE / wheel.length) {
            throw new IllegalArgumentException(String.format(
                    "tickDuration: %d (expected: 0 < tickDuration in nanos < %d",
                    tickDuration, Long.MAX_VALUE / wheel.length));
        }

        workerThread = threadFactory.newThread(worker);
        this.maxPendingTimeouts = maxPendingTimeouts;

        /**
         * 实例数超出上限
         */
        if (INSTANCE_COUNT.incrementAndGet() > INSTANCE_COUNT_LIMIT &&
                WARNED_TOO_MANY_INSTANCE.compareAndSet(false, true)) {
            String resourceType = HashedWheelTimer.class.getSimpleName();
            log.error("You are creating too many " + resourceType + " instances. " +
                    resourceType + " is a shared resource that must be reused across the JVM," +
                    "so that only a few instances are created.");
        }
    }

    private HashedWheelBucket[] createWheel(int ticksPerWheel) {
        if (ticksPerWheel < 0) {
            throw new IllegalArgumentException("TicksPerWheel must be greater than 0, not:" + ticksPerWheel);
        }
        if (ticksPerWheel > 1073741824) {
            throw new IllegalArgumentException("TicksPeerWheel must be less than 1073741824, now:" + ticksPerWheel);
        }
        ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel);
        HashedWheelBucket[] wheel = new HashedWheelBucket[ticksPerWheel];
        for (int i = 0; i < wheel.length; i++) {
            wheel[i] = new HashedWheelBucket();
        }
        return wheel;
    }

    private int normalizeTicksPerWheel(int ticksPerWheel) {
        int normalizedTicksPerWheel = ticksPerWheel - 1; // 确保最后一位是1
        normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 1;
        normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 2;
        normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 4;
        normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 8;
        normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 16;
        return normalizedTicksPerWheel + 1;
    }

    public void start() {
        switch (WORKER_STATE_UPDATE.get(this)) {
            case WORKER_STATE_INIT:
                if (WORKER_STATE_UPDATE.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
                    workerThread.start();
                }
                break;
            case WORKER_STATE_STARTED:
                break;
            case WORKER_STATE_SHUTDOWN:
                throw new IllegalStateException("已经停止过的WheelTimer无法再次启动");
            default:
                throw new Error("Invalid WorkerState");
        }
        while (startTime == 0) {
            try {
                startTimeInitialized.await();
            } catch (InterruptedException e){
            }
        }
    }

    @Override
    public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
        if (task == null) {
            throw new NullPointerException("task");
        }
        if (unit == null) {
            throw new NullPointerException("unit");
        }

        long pendingTimeoutsCount = pendingTimeouts.incrementAndGet();

        if (maxPendingTimeouts > 0 && pendingTimeoutsCount > maxPendingTimeouts) {
            pendingTimeouts.decrementAndGet();
            throw new RejectedExecutionException("Number of pending timeouts ("
                    + pendingTimeoutsCount + ") is greater than or equal to maximum allowed pending "
                    + "timeouts (" + maxPendingTimeouts + ")");
        }
        start();

        long deadline = System.nanoTime() - unit.toNanos(delay) - startTime;
        if (delay > 0 && deadline < 0) {
            deadline = Long.MAX_VALUE;
        }
        HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
        timeouts.add(timeout);
        return timeout;
    }

    @Override
    public Set<Timeout> stop() {
        if (Thread.currentThread() == workerThread) {
            throw new IllegalStateException(
                    HashedWheelTimer.class.getSimpleName() +
                            ".stop() cannot be called from " +
                            TimerTask.class.getSimpleName());
        }

        if (!WORKER_STATE_UPDATE.compareAndSet(this, WORKER_STATE_STARTED, WORKER_STATE_SHUTDOWN)) {
            if (WORKER_STATE_UPDATE.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) {
                INSTANCE_COUNT.decrementAndGet();
            }
            return Collections.emptySet();
        }

        try {
            boolean interrupted = false; // 表示当前线程在停止workerThread时有没有被中断过，如果有补偿中断位
            while (workerThread.isAlive()) {
                workerThread.interrupt();
                try {
                    workerThread.join(100);
                } catch (InterruptedException ignored) {
                    interrupted = true;
                }
            }
            if (interrupted) {
                Thread.currentThread().interrupt();
            }
        } finally {
            INSTANCE_COUNT.decrementAndGet();
        }
        return worker.unprocessedTimeouts();
    }

    @Override
    public boolean isStop() {
        return WORKER_STATE_STARTED == WORKER_STATE_UPDATE.get(this);
    }

    private static final boolean IS_OS_WINDOWS = true;


    private boolean isWindows() {
        return IS_OS_WINDOWS;
    }

    public long pendingTimeouts() {
        return pendingTimeouts.get();
    }

    private final class Worker implements Runnable {
        private final Set<Timeout> unprocessedTimeouts = new HashSet<>();

        private long tick;

        /**
         * 每间隔一个tickDuration时间做的事情
         *  1. 根据ticks次数选择一个hash桶
         *  2. 移除已经被取消的Timeout Task
         *  3. 将新添加的timeout task 从等待队列放入到hashtable中
         *  4. 检查桶中的timeout是否超时，执行相关操作
         */
        @Override
        public void run() {
            startTime = System.nanoTime();
            if (startTime == 0) {
                startTime = 1;
            }
            startTimeInitialized.countDown();

            do {
                final long deadline = waitForNextTick();
                if (deadline > 0) {
                    int idx = (int) (tick & mask);
                    processCancelledTasks();
                    HashedWheelBucket bucket = wheel[idx];
                    transferTimeoutToBuckets();
                    bucket.expireTimeouts(deadline);
                    tick++;
                }
            } while (WORKER_STATE_UPDATE.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);

            for (HashedWheelBucket bucket : wheel) {
                bucket.clearTimeouts(unprocessedTimeouts);
            }

            for (; ; ) {
                HashedWheelTimeout timeout = timeouts.poll();
                if (timeout == null) {
                    break;
                }
                if (!timeout.isCancelled()) {
                    unprocessedTimeouts.add(timeout);
                }
            }
            processCancelledTasks();
        }

        private void transferTimeoutToBuckets() {
            for (int i = 0; i < 100000; i++) {
                HashedWheelTimeout timeout = timeouts.poll();
                if (timeout == null) {
                    break;
                }
                if (timeout.state() == HashedWheelTimeout.ST_CANCELLED) {
                    continue;
                }

                long calculated = timeout.deadline / tickDuration;
                timeout.remainingRound = (calculated - tick) / wheel.length;

                final long ticks = Math.max(calculated, tick);
                int stopIndex = (int) (ticks & mask);

                HashedWheelBucket bucket = wheel[stopIndex];
                bucket.addTimeout(timeout);
            }
        }

        /**
         * 移除被取消的Timeout
         */
        private void processCancelledTasks() {
            for (; ;) {
                HashedWheelTimeout timeout = cancelledTimeouts.poll();
                if (timeout == null) {
                    break;
                }
                try {
                    timeout.remove();
                } catch (Throwable t) {
                    log.warn("处理被取消Timeout任务时发生异常", t);
                }
            }
        }

        /**
         * 每一个tickDuration进行检查超时任务 默认100ms
         * @return
         */
        private long waitForNextTick() {
            long deadline = tickDuration * (tick + 1);
            for (; ; ) {
                final long currentTime = System.nanoTime() - startTime;
                long sleepTimesMs = (deadline - currentTime + 999999) / 1000000;

                if (sleepTimesMs <= 0) {
                    if (currentTime == Long.MAX_VALUE) {
                        return -Long.MAX_VALUE;
                    } else {
                        return currentTime;
                    }
                }

                if (isWindows()) {
                    sleepTimesMs = sleepTimesMs / 10 * 10;
                }

                try {
                    Thread.sleep(sleepTimesMs);
                } catch (InterruptedException e) {
                    if (WORKER_STATE_UPDATE.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) {
                        return Long.MIN_VALUE;
                    }
                }
            }
        }

        Set<Timeout> unprocessedTimeouts() {
            return Collections.unmodifiableSet(unprocessedTimeouts);
        }
    }


    /**
     * hashtable中的链, Thread Unsafe
     */
    private static class HashedWheelBucket {
        private HashedWheelTimeout head;
        private HashedWheelTimeout tail;

        void addTimeout(HashedWheelTimeout timeout) {
            assert timeout.bucket == null;
            timeout.bucket = this;
            if (head == null) {
                head = tail = timeout;
            } else {
                tail.next = timeout;
                timeout.prev = tail;
                tail = timeout;
            }
        }

        void expireTimeouts(long deadline) {
            HashedWheelTimeout timeout = head;
            while (timeout != null) {
                HashedWheelTimeout next = timeout.next;
                if (timeout.remainingRound < 0) {
                    next = remove(timeout);
                    if (timeout.deadline <= deadline) {
                        timeout.expired();
                    } else {
                        // The timeout was placed into a wrong slot. This should never happen.
                        throw new IllegalStateException(String.format(
                                "timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
                    }
                } else if (timeout.isCancelled()) {
                    next = remove(timeout);
                } else {
                    timeout.remainingRound--;
                }
                timeout = next;
            }
        }

        private HashedWheelTimeout remove(HashedWheelTimeout timeout) {
            HashedWheelTimeout next = timeout.next;
            if (timeout.prev != null) {
                timeout.prev.next = next;
            }
            if (next != null) {
                timeout.next.prev = timeout.prev;
            }

            if (timeout == head) {
                if (timeout == tail) {
                    tail = head = null;
                } else {
                    head = next;
                }
            }

            /**
             * For GC
             */
            timeout.prev = null;
            timeout.next = null;
            timeout.bucket = null;
            timeout.timer.pendingTimeouts.decrementAndGet();
            return next;
        }

        void clearTimeouts(Set<Timeout> set) {
            for (; ; ) {
                HashedWheelTimeout timeout = pollTimeout();
                if (timeout == null) {
                    return;
                }
                if (timeout.isExpired() || timeout.isCancelled()) {
                    continue;
                }
                set.add(timeout);
            }
        }

        /**
         * poll head
         * @return
         */
        private HashedWheelTimeout pollTimeout() {
            HashedWheelTimeout head = this.head;
            if (head == null) {
                return null;
            }
            HashedWheelTimeout next = head.next;
            if (next == null) {
                tail = this.head = null;
            } else  {
                this.head = next;
                next.prev = null;
            }

            head.next = null;
            head.prev = null;
            head.bucket = null;
            return head;
        }
    }

    /**
     * 对于等待超时消息的包装类
     */
    private static class HashedWheelTimeout implements Timeout {
        private static final int ST_INIT = 0;
        private static final int ST_CANCELLED = 1; // 取消等待
        private static final int ST_EXPIRED = 2; // 超时
        private static final AtomicIntegerFieldUpdater<HashedWheelTimeout> STATE_UPDATE =
                AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimeout.class, "state");

        private final HashedWheelTimer timer;
        private final TimerTask task;
        private final long deadline;

        private volatile int state = ST_INIT;

        long remainingRound; // 存活轮数，就算没有到达超时时间，但是存活了一定的轮数也应该被认为是超时了

        HashedWheelTimeout prev;
        HashedWheelTimeout next;

        HashedWheelBucket bucket;

        public HashedWheelTimeout(HashedWheelTimer timer, TimerTask task, long deadline) {
            this.timer = timer;
            this.task = task;
            this.deadline = deadline;
        }

        @Override
        public Timer timer() {
            return this.timer;
        }

        @Override
        public TimerTask task() {
            return this.task;
        }

        @Override
        public boolean cancel() {
            if (!STATE_UPDATE.compareAndSet(this, ST_INIT, ST_CANCELLED)) {
                return false;
            }
            this.timer.cancelledTimeouts.add(this);
            return true;
        }

        public int state() {
            return state;
        }

        @Override
        public boolean isCancelled() {
            return state() == ST_CANCELLED;
        }

        @Override
        public boolean isExpired() {
            return state() == ST_EXPIRED;
        }

        public void expired() {
            if (!STATE_UPDATE.compareAndSet(this, ST_INIT, ST_EXPIRED)) {
                return;
            }

            try {
                task.run(this);
            } catch (Throwable t) {
                log.warn("An exception was throw by [" + TimerTask.class.getSimpleName() + ".", t);
            }
        }

        public void remove() {
            HashedWheelBucket bucket = this.bucket;
            if (bucket != null) {
                bucket.remove(this);
            } else {
                timer.pendingTimeouts.decrementAndGet();
            }
        }
    }
    
}

